package com.ljw.hbase;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.*;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;


/**
 * Created by admin on 2017/11/4.
 */
public class HbaseApp {
    Connection connect=null;
    Table table=null;
    Admin admin=null;
    public void startUp() throws Exception{
        Configuration configuration=new Configuration();
        configuration.set("hbase.rootdir","hdfs://192.168.170.223:9000/hbase");
        configuration.set("hbase.cluster.distributed","true");
        configuration.set("hbase.zookeeper.quorum","192.168.170.223:2181");
        connect= ConnectionFactory.createConnection(configuration);
           admin= connect.getAdmin();

    }
    /*查询所有的表*/
    public void queryTables() throws IOException {
       HTableDescriptor[] tables= admin.listTables();
        for (HTableDescriptor htabe:tables
             ) {
            String name=htabe.getTableName().getNameAsString();
            System.out.println(name);
            HColumnDescriptor[] cf=  htabe.getColumnFamilies();
            for (HColumnDescriptor columnDescriptor:cf) {
                System.out.println(columnDescriptor.getNameAsString());
            }

        }

    }
    /*创建表*/
    public void createTable(String tableName,String columnFamily) throws Exception{
        if(!admin.tableExists(TableName.valueOf(tableName))){
             HTableDescriptor hTableDescriptor=new HTableDescriptor(TableName.valueOf(tableName));
             hTableDescriptor.addFamily(new HColumnDescriptor(columnFamily));
            admin.createTable(hTableDescriptor);
        }else{
            System.out.println(tableName+"已经存在！！！！");
        }
    }
//    插入多条数据记录
    public void putList() throws Exception{
        List<Put> list = new ArrayList<Put>();
        table =connect.getTable(TableName.valueOf("member"));
        Put put1=new Put(Bytes.toBytes("zhangsan"));
        put1.addColumn(Bytes.toBytes("info"),Bytes.toBytes("age"),Bytes.toBytes(40));
        Put put2=new Put(Bytes.toBytes("Lisi"));
        put1.addColumn(Bytes.toBytes("info"),Bytes.toBytes("age"),Bytes.toBytes(40));
        Put put3=new Put(Bytes.toBytes("lisi1"));
        put1.addColumn(Bytes.toBytes("info"),Bytes.toBytes("age"),Bytes.toBytes(40));
        list.add(put1);
        list.add(put2);
        list.add(put3);
        table.put(list);

    }
    /*插入单条数据*/
    public  void putOne() throws Exception{
//        获得表
        table =connect.getTable(TableName.valueOf("member"));
//        获得put对象，添加主键数据
        Put put = new Put(Bytes.toBytes("zhangsan"));
//        数据
        put.addColumn(Bytes.toBytes("ifo"),Bytes.toBytes("age"),Bytes.toBytes("40"));
        table.put(put);

    }
//    修改数据
    public  void update() throws Exception{
        //        获得表
        table =connect.getTable(TableName.valueOf("member"));
//        获得put对象，添加主键数据
        Put put = new Put(Bytes.toBytes("zhangsan"));
//        数据
        put.addColumn(Bytes.toBytes("ifo"),Bytes.toBytes("age"),Bytes.toBytes("40"));
        table.put(put);

    }

    /**
     * 全表扫描
     * @throws Exception
     */
    public void scan01(String tableName) throws Exception{
        table=connect.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        ResultScanner rs=table.getScanner(scan);
        for (Result r:rs) {
            printResult(r);
        }

    }

    /**
     * 返回大于等于rowkey的所有记录。
     * @param tableName
     * @throws Exception
     */
    public void scan02(String tableName,String startRow) throws Exception{
        table=connect.getTable(TableName.valueOf(tableName));
//        public Scan(byte [] startRow, Filter filter) {
//         Scan(byte [] startRow)
        Scan scan = new Scan(Bytes.toBytes(startRow));
        ResultScanner rs=table.getScanner(scan);
        for (Result r:rs) {
            printResult(r);
        }

    }

    /**
     * 根据startRow和stopRow [startRow,stopRow)
     * @param tableName
     * @param startRow
     * @throws Exception
     */
    public void scan03(String tableName,String startRow,String stopRow) throws Exception{
        table=connect.getTable(TableName.valueOf(tableName));
//        public Scan(byte [] startRow, Filter filter) {
//        Scan(byte [] startRow, byte [] stopRow)
        Scan scan = new Scan(Bytes.toBytes(startRow),Bytes.toBytes(stopRow));
        ResultScanner rs=table.getScanner(scan);
        for (Result r:rs) {
            printResult(r);
        }

    }

    /**
     * 根据指定cf和column/qualifier
     * @param tableName
     *
     * @throws Exception
     */
    public void scan04(String tableName,String cf,String qualifier) throws Exception{
        table=connect.getTable(TableName.valueOf(tableName));
//        =================================================================
//        public Scan(byte [] startRow, Filter filter) {
//        ResultScanner getScanner(byte[] family) throws IOException;
//        ResultScanner rs=table.getScanner(Bytes.toBytes(cf));


//        ==============================================================
//         ResultScanner getScanner(byte[] family, byte[] qualifier) throws IOException;
       ResultScanner rs=table.getScanner(Bytes.toBytes(cf),Bytes.toBytes(qualifier));
        for (Result r:rs) {
            printResult(r);
        }

    }

    /**
     * 根据rowkey模糊查询过滤（正则）
     * @throws Exception
     */
    public  void filter01(String tableName) throws Exception{
        table=connect.getTable(TableName.valueOf(tableName));
        /**
         * ^:开头
         * $:结尾
         * *:任意多个字符
         * _:任意一个字符
         */
        String regex="^*";
        Scan scan = new Scan();
        Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL,new RegexStringComparator(regex));
        scan.setFilter(filter);
        ResultScanner rs=table.getScanner(scan);
        for (Result r:rs) {
            printResult(r);
        }
    }

    /**
     * 根据rowkey模糊查询过滤:前缀过滤
     * @param tableName
     * @throws Exception
     */
    public  void filter02(String tableName) throws Exception{
        table=connect.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        Filter filter = new PrefixFilter(Bytes.toBytes("l"));
        scan.setFilter(filter);
        ResultScanner rs=table.getScanner(scan);
        for (Result r:rs) {
            printResult(r);
        }
    }

    /**
     * 多条件组合过滤
     * @param tableName
     * @throws Exception
     */
    public  void filter03(String tableName) throws Exception{
        table=connect.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
//        public FilterList(final Filter... rowFilters) {
//        public FilterList(final Operator operator)
// Operator：操作符/** !AND */MUST_PASS_ALL, /** !OR */MUST_PASS_ONE
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ONE);
        Filter filter1 = new PrefixFilter(Bytes.toBytes("l"));
        Filter filter2 = new PrefixFilter(Bytes.toBytes("z"));
        filterList.addFilter(filter1);
        filterList.addFilter(filter2);
        scan.setFilter(filterList);
        ResultScanner rs=table.getScanner(scan);
        for (Result r:rs) {
            printResult(r);
        }
    }

    /**
     * 按照rowkey删除
     * @throws Exception
     */
    public  void delete01(String tableName,String rowkey) throws Exception{
        table=connect.getTable(TableName.valueOf(tableName));
        /*
        public Delete(byte [] row)
        Delete(byte [] row, long timestamp)
         */
        Delete delete=new Delete(Bytes.toBytes(rowkey));
        table.delete(delete);
//        删除多条rowkey   delete(List<Delete> deletes)


    }

    /**
     * 删除指定qualifier
     * @param tableName
     * @param rowkey
     * @throws Exception
     */
    public  void delete02(String tableName,String rowkey) throws Exception{
        table=connect.getTable(TableName.valueOf(tableName));
        /*
        public Delete(byte [] row)
        Delete(byte [] row, long timestamp)
         */
        Delete delete=new Delete(Bytes.toBytes(rowkey));

        String cf="cf";
        String qualifier="q";
        //        addColumn(final byte [] family, final byte [] qualifier)
//         public Delete addColumn(byte [] family, byte [] qualifier, long timestamp) {
//        删除最新的cell 信息
//        delete.addColumn(Bytes.toBytes(cf),Bytes.toBytes(qualifier));
//        删除该cell 的所有版本

        delete.addColumns(Bytes.toBytes(cf),Bytes.toBytes(qualifier));
//        ==========================================================
//        删除列组
        delete.addFamily(Bytes.toBytes(cf));

//        ==========================================================
        table.delete(delete);
//        删除多条rowkey   delete(List<Delete> deletes)
    }
    /**
     * 删除表
     */
    public void deleteTable(String tableName)throws  Exception{

        if (admin.tableExists(TableName.valueOf(tableName))){
            admin.disableTable(TableName.valueOf(tableName));
            admin.deleteTable(TableName.valueOf(tableName));
            System.out.println("删除成功！！！");


        }else{
            System.out.println(tableName+"不存在！！");
        }    }
    /**
     * 根据rowkey来获取指定记录
     * @throws Exception
     */
    public  void get01(String tableName,String rowKey) throws Exception {
        table = connect.getTable(TableName.valueOf(tableName));
        Get get = new Get(rowKey.getBytes());
        Result result = table.get(get);
        for (Cell cell : result.rawCells()) {
            System.out.println("表的主键名：" + Bytes.toString(result.getRow()));
            System.out.println("表的单元格的列组名：" + CellUtil.cloneFamily(cell));
            System.out.println("单元格的列名：" + CellUtil.cloneQualifier(cell));
            System.out.println("单元格的值：" + CellUtil.cloneValue(cell));


        }
    }
        /**
         * 根据rowkey和cf的qualify来获取指定记录
         * @throws Exception
         */
    public  void get02(String tableName,String rowKey,String cf,String columnName) throws Exception{
        table=connect.getTable(TableName.valueOf(tableName));
        Get get = new Get(rowKey.getBytes());
//        按照特定的列组查找数据
//        get.addFamily(Bytes.toBytes(cf));
        get.addColumn(Bytes.toBytes(cf),Bytes.toBytes(columnName));
        Result result= table.get(get);
//        获得查找的列的值
        System.out.println(result.getValue(Bytes.toBytes(cf),Bytes.toBytes(columnName)));
       /* for (Cell cell: result.rawCells()) {
            System.out.println("表的主键名："+Bytes.toString(result.getRow()));
            System.out.println("表的单元格的列组名："+ CellUtil.cloneFamily(cell));
            System.out.println("单元格的列名："+CellUtil.cloneQualifier(cell));
            System.out.println("单元格的值："+CellUtil.cloneValue(cell));


        }*/


    }

    /**
     * 这是一个工具方法。
     * @param result
     * @throws Exception
     */
    public void printResult(Result result) throws Exception{
        for (Cell cell: result.rawCells()) {
            System.out.println("表的主键名："+Bytes.toString(result.getRow()));
            System.out.println("表的单元格的列组名："+ CellUtil.cloneFamily(cell));
            System.out.println("单元格的列名："+CellUtil.cloneQualifier(cell));
            System.out.println("单元格的值："+CellUtil.cloneValue(cell));


        }
    }
    public void stop() throws Exception{
        connect.close();
    }

}
